package com.zy.rpc.common.discovery.impl;

import com.zy.rpc.common.balance.LoadBalance;
import com.zy.rpc.common.balance.LoadBalanceFactory;
import com.zy.rpc.common.discovery.DiscoveryService;
import com.zy.rpc.common.serviceinfo.ServiceInfo;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.springframework.util.CollectionUtils;

import java.util.Collection;
import java.util.Collections;
import java.util.stream.Collectors;

/**
 * @author zy
 * @date 2023/5/5 10:41
 */
public class ZookeeperDiscoveryService implements DiscoveryService {

    public static  final int BASE_SLEEP_TIME_MS=1000;

    public static final  int MAX_PETRIES=3;
    private LoadBalance loadBalance;

    private CuratorFramework curatorFramework;

    private ServiceDiscovery<ServiceInfo> serviceDiscovery;



    public ZookeeperDiscoveryService(String registryAddr){
        this.loadBalance= LoadBalanceFactory.getDefaultLoadBalance();
        try {
            curatorFramework= CuratorFrameworkFactory.newClient(registryAddr,new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS,MAX_PETRIES));
            curatorFramework.start();
            JsonInstanceSerializer<ServiceInfo> jsonInstanceSerializer = new JsonInstanceSerializer<>(ServiceInfo.class);
            serviceDiscovery= ServiceDiscoveryBuilder.builder(ServiceInfo.class)
                    .client(curatorFramework)
                    .serializer(jsonInstanceSerializer)
                    .basePath("/services")
                    .build();
            serviceDiscovery.start();
        } catch (Exception e) {
            throw new RuntimeException("client run error");
        }
    }


    @Override
    public ServiceInfo discovery(String serviceName) throws Exception {
        Collection<ServiceInstance<ServiceInfo>> serviceInstances = serviceDiscovery.queryForInstances(serviceName);
        return CollectionUtils.isEmpty(serviceInstances)? null:loadBalance.choose(serviceInstances.stream().map(ServiceInstance::getPayload).collect(Collectors.toList()));

    }
}
